package org.dreamwork.cai.core.impl;

import org.dreamwork.cai.core.IHeartbeatManager;
import org.dreamwork.cai.io.IRouteBroadcaster;
import org.dreamwork.cai.proxy.IServiceResolver;
import org.dreamwork.cai.srf.ServiceRouteFrame;
import org.dreamwork.cai.srf.ServiceRouteTable;
import org.dreamwork.cai.tools.Const;
import org.dreamwork.concurrent.Looper;
import org.dreamwork.concurrent.broadcast.ILocalBroadcastReceiver;
import org.dreamwork.concurrent.broadcast.ILocalBroadcastService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

public class HeartbeatManager implements IHeartbeatManager {
    private static final String LOOP_NAME = "cai.HeartbeatManager";

    private final Logger logger = LoggerFactory.getLogger (HeartbeatManager.class);

    /** 全局路由表 */
    private ServiceRouteTable table;

    /** 扫描间隔， 单位: 毫秒 */
    private int scanInterval;

    /** 心跳发送间隔，单位: 毫秒 */
    private int sendInterval;

    /** 心跳检测超时时长，单位: 毫秒 */
    private long timeout;

    /** 路由广播器 */
    private IRouteBroadcaster broadcaster;

    /** 本地服务查找器 */
    private IServiceResolver resolver;

    private final Map<String, Wrapper> hosts = Collections.synchronizedMap (new HashMap<> ());

    private boolean running = true;

    private ScheduledExecutorService executor;

    private ScheduledFuture<?> future;

    /** 本地广播的监听器，当外部节点发出 withdrawn 消息时，从本地心跳表中删除该节点 */
    private ILocalBroadcastReceiver receiver;

    public void setTable (ServiceRouteTable table) {
        this.table = table;
    }

    public void setScanInterval (int scanInterval) {
        this.scanInterval = scanInterval;
    }

    public void setSendInterval (int sendInterval) {
        this.sendInterval = sendInterval;
    }

    /**
     * 超时间隔，单位: 秒. 如果某个节点在这个时间间隔内无心跳到达，则判定为节点拖网。将自动删除该节点的路由帧
     *
     * @param timeout 超时时间
     */
    public void setTimeout (int timeout) {
        this.timeout = TimeUnit.SECONDS.toMillis (timeout);
    }

    /**
     * 设置路由广播器
     * @param broadcaster 路由广播器
     */
    public void setBroadcaster (IRouteBroadcaster broadcaster) {
        this.broadcaster = broadcaster;
    }

    public void setResolver (IServiceResolver resolver) {
        this.resolver = resolver;
    }

    public void start () {
        // 启动心跳扫描线程
        Looper.create (LOOP_NAME, 1);
        Looper.runInLoop (LOOP_NAME, () -> {
            final Logger logger = LoggerFactory.getLogger (getClass ());
            if (logger.isTraceEnabled ()) {
                logger.trace ("starting heartbeat scan thread ...");
            }
            Set<String> copy = new HashSet<> ();

            while (running) {
                while (hosts.isEmpty () && running) {
                    synchronized (LOOP_NAME) {
                        try {
                            LOOP_NAME.wait (60000);
                        } catch (InterruptedException e) {
                            e.printStackTrace ();
                        }
                    }
                }

                try {
                    for (Map.Entry<String, Wrapper> e : hosts.entrySet ()) {
                        if (System.currentTimeMillis () - e.getValue ().timestamp > timeout) {
                            String name = e.getKey ();
                            copy.add (name);
                            if (logger.isTraceEnabled ()) {
                                logger.trace ("host [{}] timed out, destroy it!", name);
                            }
                        }
                    }

/*
                    if (logger.isTraceEnabled ()) {
                        logger.trace ("scanning round complete, the namespaces needs to cleanup is: {}", copy);
                    }
*/
                    if (!copy.isEmpty ()) {
                        for (String host : copy) {
                            hosts.remove (host);
                            String[] a = host.split (":");
                            String ip = a[0];
                            int udpPort = Integer.parseInt (a[1]);
                            table.removeByHost (ip, udpPort);
                        }
                        if (logger.isTraceEnabled ()) {
                            logger.trace ("nodes [{}] are cleared", copy);
                        }
                    }
                } finally {
                    copy.clear ();
                }

                try {
                    Thread.sleep (scanInterval);
                } catch (InterruptedException e) {
                    e.printStackTrace ();
                }
            }
        });

        // 启动定期心跳发送线程
        if (sendInterval > 0) {
            executor = Executors.newSingleThreadScheduledExecutor ();
            future = executor.scheduleAtFixedRate (() -> {
                try {
                    broadcaster.heartbeat ();
                } catch (IOException ex) {
                    ex.printStackTrace ();
                }
            }, 0, sendInterval, TimeUnit.MILLISECONDS);
        }

        // 注册本地广播监听器
        ILocalBroadcastService service = resolver.get (ILocalBroadcastService.class);
        service.register (Const.Category.CAT_SRF_EVENT, receiver = message -> {
            if (message.what == Const.SRF.WITHDRAWN) {
                if (logger.isTraceEnabled ()) {
                    logger.trace ("received a withdrawn message, remove the srf from cache");
                }
                ServiceRouteFrame frame = (ServiceRouteFrame) message.arg;
                if (frame != null) {
                    hosts.remove (frame.host);
                    if (logger.isTraceEnabled ()) {
                        logger.trace ("srf {} removed", frame.host);
                    }
                }
            }
        });
    }

    public void shutdown () {
        running = false;
        // 删除本地广播监听器
        if (receiver != null) {
            ILocalBroadcastService service = resolver.get (ILocalBroadcastService.class);
            service.unregister (Const.Category.CAT_SRF_EVENT, receiver);
        }

        if (future != null) {
            future.cancel (true);
        }
        if (executor != null) {
            executor.shutdownNow ();
        }
        synchronized (LOOP_NAME) {
            LOOP_NAME.notifyAll ();
        }
    }

    @Override
    public void touch (String host, int port) {
        String name = host + ":" + port;
        Wrapper wrapper = hosts.computeIfAbsent (name, key -> {
            // 能进这里的，一定是在心跳表中不存在，发送本地广播：节点启动
            ILocalBroadcastService service = resolver.get (ILocalBroadcastService.class);
            service.broadcast (Const.Category.CAT_ROUTE_NODE, Const.RouteEvent.STARTUP, name);
            return new Wrapper (name);
        });
        wrapper.timestamp = System.currentTimeMillis ();
        synchronized (LOOP_NAME) {
            LOOP_NAME.notifyAll ();
        }
    }

/*
    public void touch (String host) {
        Wrapper wrapper = hosts.computeIfAbsent (host, key -> {
            // 能进这里的，一定是在心跳表中不存在，发送本地广播：节点启动
            ILocalBroadcastService service = resolver.get (ILocalBroadcastService.class);
            service.broadcast (Const.Category.CAT_ROUTE_NODE, Const.RouteEvent.STARTUP, host);
            return new Wrapper (host);
        });
        wrapper.timestamp = System.currentTimeMillis ();
        synchronized (LOOP_NAME) {
            LOOP_NAME.notifyAll ();
        }
    }
*/

    @Override
    public void remove (String host) {
        hosts.remove (host);
        if (logger.isTraceEnabled ()) {
            logger.trace ("node [{}] removed", host);
        }
    }

    private final static class Wrapper {
        String host;
        long timestamp;

        Wrapper (String host) {
            this.host = host;
        }
    }
}